package com.intct.flink.window;

import com.intct.hbase.bean.WindowBean;
import org.apache.commons.lang3.time.DateFormatUtils;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author gufg
 * @since 2025-06-27 14:26
 */
public class WindowProcessDemo {
    public static void main(String[] args) throws Exception {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度
        env.setParallelism(1);

        // 从socket获取数据
        DataStreamSource<String> sourceDS = env.socketTextStream("jd-node", 8888);

        // Map转换(将字符转换为实体Bean)
        SingleOutputStreamOperator<WindowBean> mapDS = sourceDS.map(sources -> {
            String[] sourceList = sources.split(",");
            return WindowBean.builder()
                    .name(sourceList[0])
                    .ts(Long.valueOf(sourceList[1]))
                    .vc(Integer.valueOf(sourceList[2]))
                    .build();
        });

        // 按照指定Key分区
        KeyedStream<WindowBean, String> keyByDS = mapDS.keyBy(key -> key.getName());

        // 窗口-- 分配器(滚动窗口) 10秒一个窗口
        WindowedStream<WindowBean, String, TimeWindow> windowDS =
                keyByDS.window(TumblingProcessingTimeWindows.of(Time.seconds(30)));

        /**
         * 用于在键(分组)窗口上使用上下文检索额外信息的函数的基本抽象类。
         * 参数:
         *      <IN>-输入值的类型。
         *      <0UT>-输出值的类型。
         *      <KEY>-键的类型。
         *      <W>-可以应用于此窗口函数的窗口类型。
         */
        SingleOutputStreamOperator<String> processDS = windowDS.process(new ProcessWindowFunction<WindowBean, String, String, TimeWindow>() {
            /**
             *
             * @param key 分组Key
             * @param context 上下文
             * @param elements 存在全窗口的所有数据
             * @param out 输出
             * @throws Exception
             */
            @Override
            public void process(String key, ProcessWindowFunction<WindowBean, String, String, TimeWindow>.Context context,
                                Iterable<WindowBean> elements, Collector<String> out) throws Exception {

                long count = elements.spliterator().estimateSize();

                long startTs = context.window().getStart();
                String startTime = DateFormatUtils.format(startTs, "yyyy-MM-dd HH:mm:ss");
                long endTs = context.window().getEnd();
                String endTime = DateFormatUtils.format(endTs, "yyyy-MM-dd HH:mm:ss");

                out.collect("key=" + key + "窗口[开始时间:" + startTime + " 结束时间: " + endTime + "] ;数据的条数=" + count + "数据=" + elements.toString());
            }
        });

        processDS.print();

        env.execute();
    }
}
